<?php

namespace Shen\Rocketmq5\Service;

use Hyperf\GoTask\IPC\SocketIPCSender;
use Throwable;

class ProducerService
{
    /**
     * 发送普通消息
     * @param string $topic rocketmq的topic
     * @param array $data 传递的数据
     * @param array $options 消息参数，可填写tag和keys和delay_time
     * @return array
     */
    public function pushMsg(string $topic, array $data, array $options = []): array
    {
        $res = [
            'result' => false,
            'msg_id' => '',
            'message' => 'Message sending failed.',
        ];
        try {
            $task = new SocketIPCSender("127.0.0.1:6001");
            $msg = [
                'topic' => $topic,
                'body' => [
                    "message" => json_encode($data),
                ],
                'tag' => '',
                'keys' => []
            ];

            if (isset($options['tag'])) {
                $msg['tag'] = is_string($options['tag']) ? $options['tag'] : $msg['tag'];
            }

            if (isset($options['keys'])) {
                foreach ($options['keys'] as $key) {
                    if (is_string($key)) {
                        $msg['keys'][] = $key;
                    }
                }
            }

            if (isset($options['delay_time']) && is_numeric($options['delay_time'])) {
                $msg['body']['is_delay'] = true;
                if ($options['delay_time'] > time()) {
                    $msg['body']['delay_time'] = (int)$options['delay_time'];
                }
            }

            $result = $task->call('Producer.SendNormalMsg', $msg);

            $res["result"] = $result["result"] ?? false;
            if ($res["result"]) {
                $res["msg_id"] = $result["msgId"] ?? "";
                $res["message"] = $result["message"] ?? "Message sent successfully.";
            } else {
                $res["message"] = $result["message"] ?? $res["message"];
            }

            return $res;
        } catch (Throwable $e) {
            $res["message"] = $e->getMessage();
            return $res;
        }
    }

    /**
     * @param string $topic
     * @param array $data
     * @param int $delay
     * @param array $options
     * @return array
     */
    public function pullDelayMsg(string $topic, array $data, int $delay, array $options = []): array
    {
        /**
         * 在broker中执行创建topic
         * 建延时topic命令：
         * sh mqadmin updateTopic -n localhost:9876 -t <topic_name> -c <cluster_name> -a +message.type=<message_type>
         * 示例： sh mqadmin updateTopic -c DefaultCluster -t dev_delay -n rmqnamesrv:9876 -a +message.type=DELAY
         * */
        $options['delay_time'] = $delay;
        return $this->pushMsg($topic, $data, $options);
    }
}